package com.openapi.server;

import com.openapi.data.JT808PackageData;
import com.openapi.data.Session;
import com.openapi.data.SessionType;
import com.openapi.protocol.ProtocolManager;
import com.openapi.service.KafkaService;
import com.openapi.service.SessionManager;
import com.openapi.service.SessionManagerFactory;
import exsun.jt808.data.base.ComStatus;
import exsun.jt808.data.base.CommonStatus;
import exsun.jt808.data.config.CommonMsgTopic;
import exsun.jt808.data.config.jt808.Jt808ReceiveMsgTopic;
import exsun.jt808.data.upload_data.Jt808EntityBase;
import exsun.jt808.data.util.GsonUtils;
import exsun.jt808.data.util.HostUtil;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.util.ReferenceCountUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import java.util.Date;

@Component
public class TCPServerHandler extends SimpleChannelInboundHandler<JT808PackageData> { // (1)

	private final Logger logger = LoggerFactory.getLogger(getClass());

	private static ProtocolManager protocolManager;

	@Autowired
	public void setMsgProcessService(ProtocolManager protocolManager) {
		TCPServerHandler.protocolManager = protocolManager;
	}

	private static KafkaService kafkaService;

	@Autowired
	public void setKafkaService(KafkaService kafkaService) {
		TCPServerHandler.kafkaService = kafkaService;
	}



	@Override
	protected void channelRead0(ChannelHandlerContext ctx, JT808PackageData pkg) throws Exception {
		try {
			SessionManager sessionManager = SessionManagerFactory.GetSessionManager(pkg.getSessionType());
			Session session = sessionManager.findByTerminalPhone(pkg.dvo);
			if ((session == null||!Session.buildId(session.getChannel()).equals(Session.buildId(ctx.channel())))
					&& pkg.dvo != null){
				session = Session.buildSession(ctx.channel(), pkg.dvo);
				sessionManager.put(pkg.dvo, session);
				//推送设备上线
				logger.info("+++++++设备连接：{}, DevSize:{},channel：{}", pkg.getSessionType(), sessionManager.devNoSize(),  ctx.channel());
				// 通讯状态发送
				pushComStatus(pkg.dvo, CommonStatus.ONLINE);
			}
			Jt808EntityBase jt808EntityBase = protocolManager.handle(pkg, ctx.channel());
			//文件信息的数据太大，序列化和推送kafka会报错
			if(jt808EntityBase != null&&!pkg.tpc.equals(Jt808ReceiveMsgTopic.fileStream)){
				protocolManager.tMQ(jt808EntityBase);
			}
		} finally {
			release(pkg);
		}
	}

	@Override
	public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { // (4)
		logger.error("发生异常:{}", cause.getMessage());
		cause.printStackTrace();
	}

	@Override
	public void channelActive(ChannelHandlerContext ctx) throws Exception {
		logger.info("终端连接:{}", ctx.channel());
	}

	@Override
	public void channelInactive(ChannelHandlerContext ctx) throws Exception {
		final String channelId = Session.buildId(ctx.channel());
		Session session = SessionManagerFactory.removeBySessionId(channelId);
		logger.debug("终端断开连接:{}", session);
		if (session != null && session.isAuthenticated()) {
			 //推送下线信息
			pushComStatus(session.getDevNo(), CommonStatus.OFFLINE);
		}
	}

	@Override
	public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
		if (IdleStateEvent.class.isAssignableFrom(evt.getClass())) {
			IdleStateEvent event = (IdleStateEvent) evt;
			if (event.state() == IdleState.READER_IDLE) {
				String channelId = Session.buildId(ctx.channel());
				Session session =  SessionManagerFactory.removeBySessionId(channelId);
				logger.error("服务器主动断开连接:{}", session);
				ctx.close();
			}
		}
	}

	private void release(Object msg) {
		try {
			ReferenceCountUtil.release(msg);
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	/**
	 * 推送通道首次连接或断开状态
	 *
	 * @param devNo
	 * @param status
	 */
	private void pushComStatus(String devNo, CommonStatus status)
	{
		SessionManager sessionManager = SessionManagerFactory.GetSessionManager(SessionType.JT_808);
		ComStatus comStatus = new ComStatus();
		comStatus.setClientNo(devNo);
		comStatus.setTime(new Date());
		comStatus.setStatus(status);
		comStatus.setTotalClients(sessionManager.devNoSize());
		comStatus.setHostName(HostUtil.getHostName());
		String sendData = GsonUtils.getJsonFromObject(comStatus);
		kafkaService.push(CommonMsgTopic.comStatus, devNo, sendData);
	}

}